package com.cw.recommend.statistics

import com.cw.recommend.common.Runner.{RunnerConfig, runSpark}
import com.cw.recommend.common.constant._

object RatingStatistics {


  def main(args: Array[String]): Unit = {

    import com.cw.recommend.common.util.MongoDBUtil._

    implicit val runner = RunnerConfig("statistics")
    runSpark { spark =>

      val ratingDF = readMongoDB(spark, RATING_COLLECTION)


      ratingDF.createTempView("rating")

      // 商品累积评价数统计
      val productPopularityDF =
        spark.sql("select productId, count(*) count from  rating group by productId order by count")
      productPopularityDF.show()

      productPopularityDF.sinkMongoDB(PRODUCT_POPULARITY)

      // 历史商品评价数统计，分年月
      val productHistoryPopularity =
        spark.sql(
          """
select productId, yearMonth, count(*) count from
(
  select
    productId,
    score,
    cast(date_format(from_unixtime(timestamp), 'yyyyMM') as int) yearMonth
  from rating
 )  group by yearMonth, productId
 order by yearMonth desc, count desc
          """
        )

      productHistoryPopularity.show()
      productHistoryPopularity.sinkMongoDB(PRODUCT_HISTORY_POPULARITY)

      // 商品品均评价
      val productAvgScoreDF =
        spark.sql("select productId, avg(score) avgScore from rating group by productId")

      productAvgScoreDF.show()
      productAvgScoreDF.sinkMongoDB(PRODUCT_AVG_SCORE)

    }

  }

}
